package com.gxy.learn.optimisticlock.task;

import com.gxy.learn.optimisticlock.entity.SysJob;
import com.gxy.learn.optimisticlock.entity.SysJobRunLog;
import com.gxy.learn.optimisticlock.enums.TaskStateEnum;
import com.gxy.learn.optimisticlock.repository.SysJobRepository;
import com.gxy.learn.optimisticlock.repository.SysJobRunLogRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.orm.ObjectOptimisticLockingFailureException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;


/**
 * 集群多任务执行
 *
 * @author: Gao xueyong
 * Create at: 2022/3/2 下午10:25
 */
@Slf4j
@Component
public class ColonyTaskRun {

    @Autowired
    private SysJobRepository jobRepository;
    @Autowired
    private SysJobRunLogRepository sysJobRunLogRepository;

    @Value("${task.executor}")
    private String executor;

    private Executor executors = new ScheduledThreadPoolExecutor(20, new BasicThreadFactory.Builder().namingPattern("task-schedule-pool-%d").daemon(true).build());

    @Scheduled(cron = "0 0/1 * * * ?")
    public void run() {
        log.info("执行权[{}]开始执行任务！", executor);
        //查询条件
        List<SysJob> jobs = jobRepository.findByState(TaskStateEnum.DONE.getCode());
        if (CollectionUtils.isEmpty(jobs)) {
            log.info("暂无需要执行的任务！");
            return;
        }
        for (SysJob job : jobs) {
            job.setState(TaskStateEnum.RUNNING.getCode());
            Long oldVersion = job.getVersion();
            Long newVersion = job.getVersion() + 1;
            job.setUpdateTime(new Date());
            try {
                Optional<SysJob> byId = jobRepository.findById(job.getId());
                if (!byId.isPresent() || !newVersion.equals(byId.get().getVersion())) {
                    log.error("[{}]任务已被占用！", job.getJobName());
                    continue;
                }
                jobRepository.saveAndFlush(job);
//                    获取当前任务的执行权成功
                log.info("[{}]抢占[{}]任务成功，原版本号[{}],新版本号[{}]", executor, job.getJobName(), oldVersion, newVersion);
                executors.execute(new Runnable() {
                    @Override
                    public void run() {
//                            写入执行log
                        sysJobRunLogRepository.save(new SysJobRunLog(job.getJobName(), executor, new Date()));
                        try {
                            Thread.sleep(100);
                        } catch (InterruptedException e) {
                            log.error(" err ", e);
                        }
                        log.info("[{}]已完成[{}]任务的处理并已更新任务状态为[{}]！", executor, job.getJobName(), TaskStateEnum.DONE.getDesc());
                    }
                });
            } catch (Exception e) {
                if (e instanceof ObjectOptimisticLockingFailureException) {
                    log.error("[{}]抢占[{}]任务失败！", executor, job.getJobName());
                    continue;
                }
//                获取当前任务的执行权失败
                log.error("[{}]抢占[{}]任务失败！", executor, job.getJobName(), e);
            }
        }
    }
}
